package com.atguigu.project;

import com.atguigu.been.OrderEvent;
import com.atguigu.been.TxEvent;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;

/**
 * @author wky
 * @create 2021-07-16-12:51
 */
public class Project_Order {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        senv.setParallelism(1);
        //senv.setRuntimeMode(RuntimeExecutionMode.BATCH);
        DataStreamSource<String> orderSource= senv.readTextFile("src/input/OrderLog.csv");
        DataStreamSource<String> receiptSource= senv.readTextFile("src/input/ReceiptLog.csv");
        SingleOutputStreamOperator<OrderEvent> ordermap = orderSource.map(new MapFunction<String, OrderEvent>() {
            @Override
            public OrderEvent map(String s) throws Exception {
                String[] datas = s.split(",");
                return new OrderEvent(
                        Long.valueOf(datas[0]),
                        datas[1],
                        datas[2],
                        Long.valueOf(datas[3]));

            }
        });
        SingleOutputStreamOperator<TxEvent> txmap = receiptSource.map(new MapFunction<String, TxEvent>() {
            @Override
            public TxEvent map(String s) throws Exception {
                String[] datas = s.split(",");
                return new TxEvent(
                        datas[0],
                        datas[1],
                        Long.valueOf(datas[2]));
            }
        });
        ConnectedStreams<TxEvent, OrderEvent> connect = txmap.connect(ordermap);

        ConnectedStreams<TxEvent, OrderEvent> connectedStreams = connect.keyBy("txId", "txId");
        connectedStreams.process(new CoProcessFunction<TxEvent, OrderEvent, String>() {

            HashMap<String, OrderEvent> orderEventHashMap = new HashMap<>();
            HashMap<String, TxEvent> txEventHashMap = new HashMap<>();
            @Override
            public void processElement1(TxEvent txEvent, Context context, Collector<String> collector) throws Exception {
                if (orderEventHashMap.containsKey(txEvent.getTxId())){
                    System.out.println("匹配成功"+txEvent.getTxId());
                    orderEventHashMap.remove(txEvent.getTxId());
                }
                else {
                    txEventHashMap.put(txEvent.getTxId(),txEvent);
                }

            }

            @Override
            public void processElement2(OrderEvent orderEvent, Context context, Collector<String> collector) throws Exception {
                if (txEventHashMap.containsKey(orderEvent.getTxId())){
                    System.out.println("匹配成功"+orderEvent.getTxId());
                    txEventHashMap.remove(orderEvent.getTxId());
                }
                else {
                    orderEventHashMap.put(orderEvent.getTxId(),orderEvent);
                }
            }
        }).print();
        senv.execute();


    }
}
